package library.net.retrofit;

import com.google.gson.JsonParseException;

import org.reactivestreams.Publisher;

import io.reactivex.Flowable;
import io.reactivex.FlowableTransformer;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import library.common.util.LogUtils;
import library.net.retrofit.exception.RxException;
import library.net.retrofit.exception.RxExceptionHandler;
import library.net.retrofit.response.RxBaseResponse;

/**
 * @author Created by qiang.hou on 2018/11/23.
 * @version 1.0
 */
public class RxSchedulers {

    public static <T> ObservableTransformer<T, T> io_main() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io())
                        .unsubscribeOn(Schedulers.io())
                        .onErrorResumeNext(new Function<Throwable, Observable<T>>() {
                            @Override
                            public Observable<T> apply(Throwable throwable) {
                                return Observable.error(RxExceptionHandler.handleException(throwable));
                            }
                        })
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    public static <T> ObservableTransformer<T, T> newThread_main() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.newThread())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    public static <T> ObservableTransformer<T, T> newThread_io() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.newThread())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(Schedulers.io());
            }
        };
    }

    public static <T> ObservableTransformer<T, T> io_io() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(Schedulers.io());
            }
        };
    }

    /**
     * from RxBaseResponse<T> to T, deal error
     *
     * @param <T> data
     * @return ObservableTransformer
     */
    public static <T> ObservableTransformer<RxBaseResponse<T>, T> response_data() {
        return new ObservableTransformer<RxBaseResponse<T>, T>() {
            @Override
            public ObservableSource<T> apply(Observable<RxBaseResponse<T>> upstream) {
                return upstream.map(new Function<RxBaseResponse<T>, T>() {

                    @Override
                    public T apply(RxBaseResponse<T> tRxBaseResponse) {
                        if (null == tRxBaseResponse) {
                            LogUtils.e("response_data response data error......");
                            throw new JsonParseException("response data error......");
                        }

                        if (tRxBaseResponse.isSuccess()) {
                            if (null == tRxBaseResponse.getData()) {
                                LogUtils.e("response_data response data error......");
                                throw new JsonParseException("response data error......");
                            }
                        } else {
                            throw new RxException(tRxBaseResponse.getCode(), tRxBaseResponse.getMessage());
                        }

                        return tRxBaseResponse.getData();
                    }
                }).onErrorResumeNext(new Function<Throwable, Observable<T>>() {
                    @Override
                    public Observable<T> apply(Throwable throwable) {
                        return Observable.error(RxExceptionHandler.handleException(throwable));
                    }
                });
            }
        };
    }

    public static <T> FlowableTransformer<T, T> io_main_flowable() {
        return new FlowableTransformer<T, T>() {
            @Override
            public Publisher<T> apply(Flowable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    public static <T> FlowableTransformer<T, T> newThread_main_flowable() {
        return new FlowableTransformer<T, T>() {
            @Override
            public Publisher<T> apply(Flowable<T> upstream) {
                return upstream.subscribeOn(Schedulers.newThread())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    public static <T> FlowableTransformer<T, T> newThread_io_flowable() {
        return new FlowableTransformer<T, T>() {
            @Override
            public Publisher<T> apply(Flowable<T> upstream) {
                return upstream.subscribeOn(Schedulers.newThread())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(Schedulers.io());
            }
        };
    }

    public static <T> FlowableTransformer<T, T> io_io_flowable() {
        return new FlowableTransformer<T, T>() {
            @Override
            public Publisher<T> apply(Flowable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(Schedulers.io());
            }
        };
    }

    /**
     * from RxBaseResponse<T> to T, deal error
     *
     * @param <T> data
     * @return ObservableTransformer
     */
    public static <T> FlowableTransformer<RxBaseResponse<T>, T> response_data_flowable() {
        return new FlowableTransformer<RxBaseResponse<T>, T>() {
            @Override
            public Publisher<T> apply(Flowable<RxBaseResponse<T>> upstream) {
                return upstream.map(new Function<RxBaseResponse<T>, T>() {

                    @Override
                    public T apply(RxBaseResponse<T> tRxBaseResponse) throws Exception {
                        if (null == tRxBaseResponse || null == tRxBaseResponse.getData()) {
                            LogUtils.e("response_data response data error......");
                            throw new JsonParseException("response data error......");
                        }

                        return tRxBaseResponse.getData();
                    }
                }).onErrorResumeNext(new Function<Throwable, Flowable<T>>() {
                    @Override
                    public Flowable<T> apply(Throwable throwable) throws Exception {
                        return Flowable.error(RxExceptionHandler.handleException(throwable));
                    }
                });
            }
        };
    }
}
